package com.shujia.hbaseapi;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.List;

public class HbaseFilterSpecialDemo {
    private HConnection conn;
    private HBaseAdmin hAdmin;

    @Before
    public void connect() {
        try {
            //1、获取Hadoop的相关配置环境
            Configuration conf = new Configuration();

            //2、获取zookeeper的配置
            conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");
            //获取与Hbase的连接，这个连接是将来可以用户获取hbase表的
            conn = HConnectionManager.createConnection(conf);

            //将来我们要对表做DDL相关操作，而对表的操作在hbase架构中是有HMaster
            hAdmin = new HBaseAdmin(conf);

            System.out.println("建立连接成功:" + conn + ", HMaster获取成功：" + hAdmin);

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 单列值过滤器
     * SingleColumnValueFilter会返回满足条件的cell所在行的所有cell的值（即会返回一行数据）
     * <p>
     * 通过SingleColumnValueFilter与查询文科班所有学生信息
     */
    @Test
    public void SingleColumnValueFilter() {
        try {
            //获取表的实例
            HTableInterface students = conn.getTable("students");

            //创建一个正则比较器
            RegexStringComparator regexStringComparator = new RegexStringComparator("^文科.*");

            //创建单列值过滤器对象
            SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                    "info".getBytes(),
                    "clazz".getBytes(),
                    CompareFilter.CompareOp.EQUAL,
                    regexStringComparator
            );

            Scan scan = new Scan();
            scan.setFilter(singleColumnValueFilter);

            ResultScanner scanner = students.getScanner(scan);
            print(scanner);


        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 列值排除过滤器
     * 与SingleColumnValueFilter相反，会排除掉指定的列，其他的列全部返回
     * <p>
     * 通过SingleColumnValueExcludeFilter与BinaryComparator查询文科一班所有学生信息，最终不返回clazz列
     */
    @Test
    public void SingleColumnValueExcludeFilter() {
        try {
            //获取表的实例
            HTableInterface students = conn.getTable("students");

            //创建一个二进制比较器
            BinaryComparator binaryComparator = new BinaryComparator("文科一班".getBytes());

            //创建一个列值排除过滤器
            SingleColumnValueExcludeFilter singleColumnValueExcludeFilter = new SingleColumnValueExcludeFilter(
                    "info".getBytes(),
                    "clazz".getBytes(),
                    CompareFilter.CompareOp.EQUAL,
                    binaryComparator
            );

            Scan scan = new Scan();
            scan.setFilter(singleColumnValueExcludeFilter);

            ResultScanner scanner = students.getScanner(scan);
            print(scanner);


        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * rowkey前缀过滤器
     * <p>
     * 通过PrefixFilter查询以150010008开头的所有前缀的rowkey
     */
    @Test
    public void PrefixFilter() {
        try {
            //获取表的实例
            HTableInterface students = conn.getTable("students");

            //创建rowkey前缀过滤器
            PrefixFilter prefixFilter = new PrefixFilter("150010008".getBytes());
            Scan scan = new Scan();

            scan.setFilter(prefixFilter);
            ResultScanner scanner = students.getScanner(scan);
            print2(scanner);


        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    /**
     * 分页过滤器
     * 通过PageFilter查询三页的数据，每页10条
     */

    @Test
    public void PageFilter() {
        try {
            //获取表的实例
            HTableInterface students = conn.getTable("students");
            //定义要查询的页数
            int pageNum = 3;
            //定义每页的条数
            int pageSize = 10;
            Scan scan = new Scan();

            //定义一开始的行
            String current_page_start_row = "";

            for (int i = 1; i <= pageNum; i++) {
                System.out.println("====================当前是第" + i + "页===========================");
                //创建一个分页过滤器
                PageFilter pageFilter = new PageFilter(pageSize);
                scan.setFilter(pageFilter);
                ResultScanner scanner = students.getScanner(scan);
                for (Result rs : scanner) {
                    current_page_start_row = Bytes.toString(rs.getRow());
                    //告诉扫描器是从哪一行开始获取数据
                    scan.withStartRow((current_page_start_row + 0).getBytes());
                    PageFilter pageFilter1 = new PageFilter(pageSize);
                    scan.setFilter(pageFilter1);
                    //获取id
                    String id = Bytes.toString(rs.getRow());
                    String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
                    String age = Bytes.toString(rs.getValue("info".getBytes(), "age".getBytes()));
                    String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
                    String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));
                    System.out.println("学号：" + id + ",姓名：" + name + ",年龄：" + age + ",性别：" + gender + ",班级：" + clazz);

                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void baoZhuang1() throws IOException {
        HTableInterface students = conn.getTable("students");

        Filter filter1 = new RowFilter(CompareFilter.CompareOp.NOT_EQUAL,new BinaryComparator("1500100009".getBytes()));

        //不做包装
        Scan scan = new Scan();
        scan.setFilter(filter1);
        ResultScanner scanner1 = students.getScanner(scan);
        Result rs = scanner1.next();
        while (rs != null) {
            String id = Bytes.toString(rs.getRow());
            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(rs.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));

            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz + "\t");

            rs = scanner1.next();
        }

        System.out.println("-----------------------------------------------------------------------------");

        // 使用 WhileMatchFilter 进行包装
        Filter filter2 = new WhileMatchFilter(filter1);
        scan.setFilter(filter2);
        ResultScanner scanner = students.getScanner(scan);
        Result rs2 = scanner.next();
        while (rs2 != null) {
            String id = Bytes.toString(rs2.getRow());
            String name = Bytes.toString(rs2.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(rs2.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(rs2.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(rs2.getValue("info".getBytes(), "clazz".getBytes()));

            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz + "\t");

            rs2 = scanner.next();
        }
    }



    /**
     * 专门用来打印数据的方法
     */
    public void print(ResultScanner scanner) throws IOException {
        Result rs = null;
        while ((rs = scanner.next()) != null) {
            String id = Bytes.toString(rs.getRow());
            List<Cell> cells = rs.listCells();
            System.out.print("id:" + id);
            System.out.print("\t");
            for (Cell cell : cells) {
//                String s = Bytes.toString(cell.getValue());
                String col = Bytes.toString(CellUtil.cloneQualifier(cell));
                String s = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.print(col + ":" + s);
                System.out.print("\t");
            }
            System.out.println();

//            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
//            String age = Bytes.toString(rs.getValue("info".getBytes(), "age".getBytes()));
//            String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
//            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));
//            System.out.println("学号：" + id + ",姓名：" + name + ",年龄：" + age + ",性别：" + gender + ",班级：" + clazz);

        }
    }


    /**
     * 专门用来打印数据的方法
     */
    public void print2(ResultScanner scanner) throws IOException {
        Result rs = null;
        while ((rs = scanner.next()) != null) {
            String id = Bytes.toString(rs.getRow());

            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(rs.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));
            System.out.println("学号：" + id + ",姓名：" + name + ",年龄：" + age + ",性别：" + gender + ",班级：" + clazz);

        }
    }


    @After
    public void close() {
        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            System.out.println("conn连接已经关闭.....");
        }

        if (hAdmin != null) {
            try {
                hAdmin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            System.out.println("HMaster已经关闭......");
        }
    }
}
